package com.atguigu.flink.wordcount;

import com.atguigu.flink.function.MyWordCountFlatMapFunction;
import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import static org.apache.flink.api.common.typeinfo.Types.*;



/**
 * Created by Smexy on 2023/3/31
 *
        如何为接口提供实现:
            ①可以创建外部类 Implements 接口
            ②创建一个(静态)内部类  Implements 接口
                静态:  内部类.构造()
                非静态的:  new 外部类().new 内部类()

                接口的实现，是否需要使用外部类中的属性！

            ③当前接口只使用一次，可以使用匿名内部类创建
            ④接口是函数式接口，可以使用lamda表达式
 *
 */
public class Demo9_InterfaceImplDemo
{
    private static String name = "haha";

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);
        SingleOutputStreamOperator<WordCount> ds2 = ds
            //自己实现函数接口
            .flatMap(new MyWordCountFlatMapFunction2());
        ds2
            .keyBy(WordCount::getWord)
            .sum("count")
            .print();


        env.execute();

    }

    public static class MyWordCountFlatMapFunction2  implements FlatMapFunction<String, WordCount>
    {
        @Override
        public void flatMap(String line, Collector<WordCount> out) throws Exception {
            System.out.println(name);
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(new WordCount(word, 1));
            }
        }
    }
}
